package connector

import (
	"bufio"
	"context"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"sort"
	"strings"
	"sync"
	"time"
)

// CSVSourceConnector represents a CSV file source connector
type CSVSourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// file is the file handle
	file *os.File
	// reader is the CSV reader
	reader *csv.Reader
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// filePath is the file path
	filePath string
	// hasHeader indicates if the CSV file has a header
	hasHeader bool
	// delimiter is the CSV delimiter
	delimiter rune
	// headers are the CSV headers
	headers []string
	// position is the current position in the file
	position int64
	// lineCount is the total number of lines in the file
	lineCount int
}

// NewCSVSourceConnector creates a new CSV file source connector
func NewCSVSourceConnector() SourceConnector {
	return &CSVSourceConnector{
		metrics:   make(map[string]interface{}),
		hasHeader: true,
		delimiter: ',',
	}
}

// Initialize initializes the connector
func (c *CSVSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get file path
	if path, ok := config["path"].(string); ok {
		c.filePath = path
	} else {
		return fmt.Errorf("path is required")
	}

	// Get header flag
	if hasHeader, ok := config["header"].(bool); ok {
		c.hasHeader = hasHeader
	}

	// Get delimiter
	if delimStr, ok := config["delimiter"].(string); ok && len(delimStr) > 0 {
		c.delimiter = rune(delimStr[0])
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *CSVSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Open the file
	file, err := os.Open(c.filePath)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	c.file = file

	// Create a CSV reader
	reader := csv.NewReader(file)
	reader.Comma = c.delimiter
	reader.LazyQuotes = true
	reader.TrimLeadingSpace = true
	c.reader = reader

	// Count lines and read headers if needed
	lineCount := 0
	if c.hasHeader {
		// Read headers
		headers, err := reader.Read()
		if err != nil {
			file.Close()
			return fmt.Errorf("failed to read CSV headers: %w", err)
		}
		c.headers = headers
		lineCount = 1 // Start with 1 to account for header
	}

	// Count remaining lines
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		lineCount++
	}
	if err := scanner.Err(); err != nil {
		file.Close()
		return fmt.Errorf("failed to scan file: %w", err)
	}

	// Reset the file
	_, err = file.Seek(0, 0)
	if err != nil {
		file.Close()
		return fmt.Errorf("failed to reset file: %w", err)
	}

	// Recreate the reader
	reader = csv.NewReader(file)
	reader.Comma = c.delimiter
	reader.LazyQuotes = true
	reader.TrimLeadingSpace = true
	c.reader = reader

	// Skip header if needed
	if c.hasHeader {
		_, err = reader.Read()
		if err != nil {
			file.Close()
			return fmt.Errorf("failed to skip CSV header: %w", err)
		}
	}

	// Update metrics
	c.lineCount = lineCount
	if c.hasHeader {
		c.metrics["records_total"] = lineCount - 1 // Subtract header
	} else {
		c.metrics["records_total"] = lineCount
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *CSVSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Close the file
	if c.file != nil {
		err := c.file.Close()
		if err != nil {
			return fmt.Errorf("failed to close file: %w", err)
		}
		c.file = nil
		c.reader = nil
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *CSVSourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if reader is valid
	if c.reader == nil {
		return nil, fmt.Errorf("reader not initialized")
	}

	// Get batch size
	batchSize := 10
	if size, ok := c.config["batch_size"].(int); ok {
		batchSize = size
	}

	// Read records
	records := make([]*Record, 0, batchSize)
	for i := 0; i < batchSize; i++ {
		// Read a row
		row, err := c.reader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("failed to read CSV row: %w", err)
		}

		// Create record
		record, err := c.createRecordFromRow(row)
		if err != nil {
			return nil, fmt.Errorf("failed to create record: %w", err)
		}
		records = append(records, record)
	}

	// Check if no records were read
	if len(records) == 0 {
		return nil, nil
	}

	// Update metrics
	c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
	c.metrics["last_read_time"] = time.Now().Unix()

	return NewRecordBatch(records), nil
}

// Commit commits the offset
func (c *CSVSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// CSV source does not support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *CSVSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// createRecordFromRow creates a record from a CSV row
func (c *CSVSourceConnector) createRecordFromRow(row []string) (*Record, error) {
	// Create a map for the record
	data := make(map[string]interface{})

	// If we have headers, use them as keys
	if c.hasHeader && len(c.headers) > 0 {
		for i, value := range row {
			if i < len(c.headers) {
				data[c.headers[i]] = value
			}
		}
	} else {
		// Otherwise, use column indices as keys
		for i, value := range row {
			data[fmt.Sprintf("column%d", i+1)] = value
		}
	}

	// Convert to JSON
	jsonBytes, err := json.Marshal(data)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal record: %w", err)
	}

	// Create record
	record := NewRecord(nil, jsonBytes)
	record.Timestamp = time.Now()

	// Set key if available
	if len(row) > 0 {
		// Use first column as key by default
		record.Key = []byte(row[0])

		// Check if there's a specific key column
		if keyColumn, ok := c.config["key_column"].(string); ok && c.hasHeader && len(c.headers) > 0 {
			for i, header := range c.headers {
				if header == keyColumn && i < len(row) {
					record.Key = []byte(row[i])
					break
				}
			}
		}
	}

	return record, nil
}

// CSVSinkConnector represents a CSV file sink connector
type CSVSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// file is the file handle
	file *os.File
	// writer is the CSV writer
	writer *csv.Writer
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// filePath is the file path
	filePath string
	// hasHeader indicates if the CSV file should have a header
	hasHeader bool
	// delimiter is the CSV delimiter
	delimiter rune
	// headers are the CSV headers
	headers []string
	// append indicates if the file should be appended
	append bool
	// headerWritten indicates if the header has been written
	headerWritten bool
}

// NewCSVSinkConnector creates a new CSV file sink connector
func NewCSVSinkConnector() SinkConnector {
	return &CSVSinkConnector{
		metrics:   make(map[string]interface{}),
		hasHeader: true,
		delimiter: ',',
		append:    false,
	}
}

// Initialize initializes the connector
func (c *CSVSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get file path
	if path, ok := config["path"].(string); ok {
		c.filePath = path
	} else {
		return fmt.Errorf("path is required")
	}

	// Get header flag
	if hasHeader, ok := config["header"].(bool); ok {
		c.hasHeader = hasHeader
	}

	// Get delimiter
	if delimStr, ok := config["delimiter"].(string); ok && len(delimStr) > 0 {
		c.delimiter = rune(delimStr[0])
	}

	// Get headers
	if headers, ok := config["headers"].([]string); ok {
		c.headers = headers
	} else if headersStr, ok := config["headers"].(string); ok {
		c.headers = strings.Split(headersStr, ",")
		for i := range c.headers {
			c.headers[i] = strings.TrimSpace(c.headers[i])
		}
	}

	// Get append
	if append, ok := config["append"].(bool); ok {
		c.append = append
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *CSVSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Open the file
	flag := os.O_CREATE | os.O_WRONLY
	if c.append {
		flag |= os.O_APPEND
	} else {
		flag |= os.O_TRUNC
	}
	file, err := os.OpenFile(c.filePath, flag, 0644)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	c.file = file

	// Create a CSV writer
	writer := csv.NewWriter(file)
	writer.Comma = c.delimiter
	c.writer = writer

	// Check if we need to write headers
	if c.hasHeader && !c.append {
		// If headers are not provided, we'll write them when we get the first record
		if len(c.headers) > 0 {
			err = writer.Write(c.headers)
			if err != nil {
				file.Close()
				return fmt.Errorf("failed to write CSV headers: %w", err)
			}
			writer.Flush()
			c.headerWritten = true
		}
	} else if c.append {
		// If appending, assume header is already written
		c.headerWritten = true
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *CSVSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Flush the writer
	if c.writer != nil {
		c.writer.Flush()
	}

	// Close the file
	if c.file != nil {
		err := c.file.Close()
		if err != nil {
			return fmt.Errorf("failed to close file: %w", err)
		}
		c.file = nil
		c.writer = nil
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *CSVSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Check if writer is valid
	if c.writer == nil {
		return fmt.Errorf("writer not initialized")
	}

	// Process each record
	for i, record := range batch.Records {
		// Parse record value
		var data map[string]interface{}
		err := json.Unmarshal(record.Value, &data)
		if err != nil {
			return fmt.Errorf("failed to parse record value: %w", err)
		}

		// If this is the first record and we need to write headers
		if i == 0 && c.hasHeader && !c.headerWritten {
			// Extract headers from the first record if not provided
			if len(c.headers) == 0 {
				c.headers = make([]string, 0, len(data))
				for key := range data {
					c.headers = append(c.headers, key)
				}
				// Sort headers for consistency
				sort.Strings(c.headers)
			}

			// Write headers
			err = c.writer.Write(c.headers)
			if err != nil {
				return fmt.Errorf("failed to write CSV headers: %w", err)
			}
			c.headerWritten = true
		}

		// Convert record to CSV row
		row := make([]string, len(c.headers))
		for i, header := range c.headers {
			if value, ok := data[header]; ok {
				row[i] = fmt.Sprintf("%v", value)
			}
		}

		// Write row
		err = c.writer.Write(row)
		if err != nil {
			return fmt.Errorf("failed to write CSV row: %w", err)
		}
	}

	// Flush the writer
	c.writer.Flush()
	if err := c.writer.Error(); err != nil {
		return fmt.Errorf("failed to flush CSV writer: %w", err)
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *CSVSinkConnector) Flush(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if writer is valid
	if c.writer == nil {
		return fmt.Errorf("writer not initialized")
	}

	// Flush the writer
	c.writer.Flush()
	if err := c.writer.Error(); err != nil {
		return fmt.Errorf("failed to flush CSV writer: %w", err)
	}

	return nil
}

// GetMetrics gets connector metrics
func (c *CSVSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}
